package com.ibm.cps.kafka;

import java.util.ArrayList;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import com.google.common.base.Throwables;
import com.ibm.util.ErrorCode;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;

/**
 * A wrapper of Kafka producer
 * 
 * @author Hongwei Wang
 * 
 */
public class KafkaProducer {
	private Producer<String, byte[]> producer;

	/**
	 * Constructor
	 *
	 *            brokers address
	 */
	public KafkaProducer(String brokers) {
		Properties props = new Properties();

		// broker list with format: host:port
		props.put("metadata.broker.list", brokers);
		props.put("request.required.acks", "1");
		props.put("message.send.max.retries", LocalConfig.KAFKA_MAX_RETRIES);
		props.put("key.serializer.class", "kafka.serializer.StringEncoder");
		props.put("partitioner.class", "com.ibm.cps.kafka.MessagePartitioner");

		ProducerConfig config = new ProducerConfig(props);
		producer = new Producer<String, byte[]>(config);
	}

	/**
	 * Send a message to the specified topic
	 * 
	 * @param topic
	 * @param message
	 * @throws CPSException 
	 */
	public void send(String topic, byte[] message) throws CPSException {
		try {
			KeyedMessage<String, byte[]> keyedmessage = new KeyedMessage<String, byte[]>(
					topic, message);
			producer.send(keyedmessage);
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					Throwables.getStackTraceAsString(e));
		}
	}

	/**
	 * Send a message to the specified topic
	 *
	 * @param topic
	 * @param message
	 * @throws CPSException 
	 */
	public void send(String topic, long partitionKey, byte[] message)
			throws CPSException {
		try {
			KeyedMessage<String, byte[]> keyedmessage = new KeyedMessage<String, byte[]>(
					topic, partitionKey + "", message);
			producer.send(keyedmessage);
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					Throwables.getStackTraceAsString(e));
		}
	}
	/**
	 * Send a message to the specified topic,key ,partition
	 *
	 * @throws CPSException
	 */
	public void send(String topic,String key, long partitionKey, byte[] message)
			throws CPSException {
		try {
			KeyedMessage<String, byte[]> keyedmessage = new KeyedMessage<String, byte[]>(
					topic, key,partitionKey + "", message);
			producer.send(keyedmessage);
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					Throwables.getStackTraceAsString(e));
		}
	}
	/**
	 * Send messages to the specified topic
	 *
	 * @throws CPSException
	 */
	public void send(ArrayList<KeyedMessage<String, byte[]>> keyedmessages)
			throws CPSException {
		try {
			producer.send(keyedmessages);
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					Throwables.getStackTraceAsString(e));
		}
	}
	/**
	 * Close producer
	 */
	public void close() {
		producer.close();
	}

	public static void main(String[] args) throws CPSException {
		KafkaProducer producer = new KafkaProducer("9.186.107.15:9092");
		// KafkaProducer producer = new KafkaProducer("9.186.88.253:9092");
		producer.send("xxxxxxxxxxxxxxxxxxxxxx", "hello".getBytes());
	}
}
